package com.spring.nio.mysql.test;

import com.google.common.collect.Lists;
import com.spring.nio.mysql.common.MysqlPacket;
import com.spring.nio.mysql.handshake.HandshakePacket;
import com.spring.nio.util.BasicUtil;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author : pangfuzhong
 * @description mysql通信demo
 * @date : 2021/8/7 11:40
 *
 *      Step1:客户端向DB发起TCP握手。
 *      Step2:三次握手成功。与通常流程不同的是，由DB发送HandShake信息。这个Packet里面包含了MySql的能力、加密seed等信息。
 *      Step3:客户端根据HandShake包里面的加密seed对MySql登录密码进行摘要后，构造Auth认证包发送给DB。
 *      Step4:DB接收到客户端发过来的Auth包后会对密码摘要进行比对，从而确认是否能够登录。如果能，则发送Okay包返回。
 *      Step5:客户端与DB的连接至此完毕。
 *
 *      mysql通信报文组成:
 *
 *          消息长度  消息序列号  消息体
 *
 *           3byte    1byte     ...bytes
 */
@Slf4j
public class MysqlBinlogClient {

    private static final String mysqlIp = "192.168.0.103", user = "root", password = "spring";

    private static final Integer port = 3306;

    public static void main(String[] args) {
        MysqlBinlogClient client = new MysqlBinlogClient();
        client.handlerMysqlBinlog();
    }

    private void handlerMysqlBinlog() {
        this.connectMysql();
    }

    private void connectMysql() {
        try {
            InetSocketAddress inetSocketAddress = new InetSocketAddress(InetAddress.getByName(mysqlIp), port);
            SocketChannel socketChannel = SocketChannel.open();
            socketChannel.configureBlocking(Boolean.TRUE);
            Boolean connect = socketChannel.connect(inetSocketAddress);
            while (!connect) {
                if(socketChannel.finishConnect()) {
                    break;
                }
                TimeUnit.MILLISECONDS.sleep(1000);
            }

            log.info("与mysql服务端socket连接成功, clientIp = {}, connect = {}", socketChannel.getLocalAddress(), connect);

            // 接收mysql server初始握手认证协议
            List<byte[]> readByteList = Lists.newArrayList();
            int readSize = 0;
            ByteBuffer byteBuffer = ByteBuffer.allocate(2048);
            while ((readSize = socketChannel.read(byteBuffer)) > 0) {
                byteBuffer.flip();

                MysqlPacket mysqlPacket = new MysqlPacket();
                mysqlPacket.setPacketBodyLength(BasicUtil.read3BytesToInt(byteBuffer));
                mysqlPacket.setPacketIndex(BasicUtil.readByte(byteBuffer));
                mysqlPacket.setPacketBodyData(BasicUtil.readAssignBytes(byteBuffer, mysqlPacket.getPacketBodyLength()));

                log.info("mysql server handshake packet: packetLength = {}, sequenceNum = {}", mysqlPacket.getPacketBodyLength(), mysqlPacket.getPacketIndex());
                if(byteBuffer.position() == byteBuffer.limit()) {
                    // 读取完毕
                    log.info("开始解析mysql包协议内容: ---> ");
                    this.parsePacketContent(mysqlPacket.getPacketBodyData());
                    break;
                }
            }
        } catch (UnknownHostException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 解析mysql 初始握手协议服务端反馈内容
     *  1byte: 协议版本号 <---> 3byte: 服务协议版本号
     *  4byte: 服务器线程ID
     *  8byte: 挑战随机数 --> MySQL数据库用户认证采用的是挑战/应答的方式，服务器生成该挑战数并发送给客户端，由客户端进行处理并返回相应结果，然后服务器检查是否与预期的结果相同，从而完成用户认证的过
     *  1byte: 填充值  2byte: 服务器权能标志  1byte: 字符编码
     *  2byte: 服务器状态   2byte: 服务器全能标志
     * */
    private void parsePacketContent(byte[] packetContentBytes) {
        HandshakePacket handshakePacket = new HandshakePacket(packetContentBytes);
        // 1byte : 协议版本号
        handshakePacket.setProtocolVersion(handshakePacket.readOneByte());
        log.info("协议版本号: protocolVersion = {}, position = {}", handshakePacket.getProtocolVersion(), handshakePacket.getPosition());

        // ...bytes : 数据库版本信息
        handshakePacket.setServerVersion(handshakePacket.readLoopUtilZeroByte());
        log.info("服务版本信息: byte[] = {}, serverVersion = {}, position = {}", handshakePacket.getServerVersion(), new String(handshakePacket.getServerVersion()), handshakePacket.getPosition());

        // 4bytes : 连接MySQL Server启动的线程ID
        handshakePacket.setThreadId(handshakePacket.read4BytesInt());
        log.info("server线程ID: threadId = {}, position = {}", handshakePacket.getThreadId(), handshakePacket.getPosition());

        // 8bytes : 挑战随机数，用于数据库认证
        handshakePacket.setSeed1(handshakePacket.readLoopUtilZeroByte());
        log.info("挑战随机数: byte[] = {}, seed = {}, position = {}", handshakePacket.getSeed1(), BasicUtil.bytesToLong(handshakePacket.getSeed1()), handshakePacket.getPosition());

        // 1byte : 填充值(0x00)
        handshakePacket.setFillByte((byte)0);
        log.info("填充值: byte = {}, position = {}", handshakePacket.getFillByte(), handshakePacket.getPosition());

        // 2bytes : 用于与客户端协商通讯方式, 服务端能力标志
        handshakePacket.setServerCapabilities(handshakePacket.read2BytesToInt());
        log.info("服务端能力标志: serverCapabilitiesFirst = {}, position = {}", handshakePacket.getServerCapabilities(), handshakePacket.getPosition());

        // 1byte : 数据库的编码
        handshakePacket.setServerCharsetIndex(handshakePacket.readOneByte());
        log.info("数据库编码: serverCharsetIndex = {}, position = {}", handshakePacket.getServerCharsetIndex(), handshakePacket.getPosition());

        // 2bytes : 服务器状态
        handshakePacket.setServerStatus(handshakePacket.read2BytesToInt());
        log.info("服务器状态: serverStatus = {}, position = {}", handshakePacket.getServerStatus(), handshakePacket.getPosition());

        // 13bytes : 预留字节
        byte[] moveBytes = handshakePacket.move(13);
        log.info("预留字节(跳过): moveBytes = {}, position = {}", moveBytes, handshakePacket.getPosition());

        // 12bytes : 挑战随机数，用于数据库认证
        handshakePacket.setSeed2(handshakePacket.readLoopUtilZeroByte());
        log.info("挑战随机数: byte[] = {}, position = {}", handshakePacket.getSeed2(), handshakePacket.getPosition());

        // 1byte : 填充值(0x00)
        handshakePacket.setFillByte((byte) 0);
        log.info("填充值: byte = {}, position = {}", handshakePacket.getFillByte(), handshakePacket.getPosition());

        // 剩余字节
        handshakePacket.setRestOfScrambleBuff(handshakePacket.readLoopUtilZeroByte());
        log.info("剩余最后数据: byte[] = {}, rest = {}", handshakePacket.getRestOfScrambleBuff(), new String(handshakePacket.getRestOfScrambleBuff(), Charset.forName("utf8")));
    }
}
